package com.kongyz.client;

import com.kongyz.protocol.RpcRequest;
import com.kongyz.protocol.RpcResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 用于 RPC 异步调用
 */
public class RpcFuture implements Future<Object> {
    private static final Logger logger = LoggerFactory.getLogger(RpcFuture.class);

    private Sync sync;
    private RpcRequest request;
    private RpcResponse response;
    private long startTime;
    private long responseTimeThreshold=5000;

    private List<AsyncRpcCallback> pendingCallbacks=new ArrayList<AsyncRpcCallback>();
    private ReentrantLock lock=new ReentrantLock();

    public RpcFuture( RpcRequest request) {
        this.sync = new Sync();
        this.request = request;
        this.startTime = System.currentTimeMillis();
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isCancelled() {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isDone() {
        return sync.isDone();
    }

    @Override
    public Object get() throws InterruptedException, ExecutionException {
        sync.acquire(-1);
        if(this.response!=null){
            return this.response.getResult();
        }else{
            return null;
        }
    }

    @Override
    public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        boolean success=sync.tryAcquireNanos(-1,unit.toNanos(timeout));
        if(success){
            if(this.response!=null){
                return this.response.getResult();
            }else {
                return null;
            }
        }else {
            throw new RuntimeException("Timeout exception. Request id: " + this.request.getRequestId()
                    + ". Request class name: " + this.request.getClassName()
                    + ". Request method: " + this.request.getMethodName());
        }
    }

    public void done(RpcResponse response){
        this.response=response;
        sync.release(1);
        invokeCallbacks();
        long responseTime = System.currentTimeMillis() - startTime;
        if (responseTime > this.responseTimeThreshold) {
            logger.warn("服务响应时间太长。 Request id = " + response.getRequestId() + ". Response Time = " + responseTime + "ms");
        }

    }

    private void invokeCallbacks(){
        lock.lock();
        try {
            for (final AsyncRpcCallback callback:pendingCallbacks){
                runCallback(callback);
            }
        }finally {
            lock.unlock();
        }
    }

    private void runCallback(final AsyncRpcCallback callback){
        final RpcResponse res=this.response;
        RpcClient.submit(new Runnable() {
            @Override
            public void run() {
                if(!res.isError()){
                    callback.success(res.getResult());
                }else{
                    callback.fail(new RuntimeException("Response error", new Throwable(res.getError())));
                }
            }
        });

    }


    static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID=1L;

        //future status
        private final int done=1;
        private final int pending=0;

        @Override
        protected boolean tryAcquire(int arg) {
            return getState()==done;
        }

        /**
         * Attempts to set the state to reflect a release in exclusive
         * mode.
         * <p>
         * <p>This method is always invoked by the thread performing release.
         * <p>
         * <p>The default implementation throws
         * {@link UnsupportedOperationException}.
         *
         * @param arg the release argument. This value is always the one
         *            passed to a release method, or the current state value upon
         *            entry to a condition wait.  The value is otherwise
         *            uninterpreted and can represent anything you like.
         * @return {@code true} if this object is now in a fully released
         * state, so that any waiting threads may attempt to acquire;
         * and {@code false} otherwise.
         * @throws IllegalMonitorStateException  if releasing would place this
         *                                       synchronizer in an illegal state. This exception must be
         *                                       thrown in a consistent fashion for synchronization to work
         *                                       correctly.
         * @throws UnsupportedOperationException if exclusive mode is not supported
         */
        @Override
        protected boolean tryRelease(int arg) {
            if(getState()==pending){
                if(compareAndSetState(pending,done)){
                    return true;
                }else {
                    return false;
                }
            }else {
                return true;
            }
        }

        public boolean isDone(){
            getState();
            return getState()==done;
        }

    }



}
